package cn.doitedu.cn.doitedu.rtdw.etl;

import cn.doitedu.cn.doitedu.rtdw.ddl.CatalogDatabaseSwitch;
import cn.doitedu.cn.doitedu.rtdw.udfs.TimestampTrunc;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class Job2_TrafficOlapSupport {
    public static void main(String[] args) {
        // 编程环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.enableCheckpointing(5000, CheckpointingMode.EXACTLY_ONCE);
        env.getCheckpointConfig().setCheckpointStorage("file:/d:/ckpt");
        env.setParallelism(1);

        StreamTableEnvironment tenv = StreamTableEnvironment.create(env);

        // 切换 catalog和 database
        CatalogDatabaseSwitch.switchCatalogDatabase("doit40-hive", "doit40_rtdw", tenv);

        // 注册时间取整函数
        tenv.createTemporaryFunction("time_trunc", TimestampTrunc.class);

        // doris 目标表创建
        tenv.executeSql("drop table if exists tfc_overview_u_m_doris");
        tenv.executeSql(
                " create table  if not exists tfc_overview_u_m_doris (  " +
                        "     dt              DATE,                         " +
                        "     time_60m        STRING,                       " +
                        "     time_30m        STRING,                       " +
                        "     time_10m        STRING,                       " +
                        "     time_5m         STRING,                       " +
                        "     gps_province    STRING,                       " +
                        "     gps_city        STRING,                       " +
                        "     gps_region      STRING,                       " +
                        "     release_channel STRING,                       " +
                        "     device_type     STRING,                       " +
                        "     page_type       STRING,                       " +
                        "     page_service    STRING,                       " +
                        "     is_newuser      INT,                          " +
                        " 	  pv_cnt          bigint,                       " +
                        "     uv_cnt          int,                          " +
                        " 	  session_cnt     STRING                        " +
                        " ) WITH (                                          " +
                        "    'connector' = 'doris',                         " +
                        "    'fenodes' = 'doitedu:8030',                    " +
                        "    'table.identifier' = 'dws.tfc_overview_u_m',   " +
                        "    'username' = 'root',                           " +
                        "    'password' = 'root',                           " +
                        "    'sink.label-prefix' = 'doris_label-xxx',       " +
                        "    'sink.properties.columns' = 'dt,time_60m,time_30m,time_10m,time_5m,gps_province,gps_city,gps_region,release_channel,device_type,page_type,page_service,is_newuser,pv_cnt,uv_cnt,session_cnt,uv_cnt=to_bitmap(uv_cnt),session_cnt=hll_hash(session_cnt) '  " +
                        " )                                                 "
        );

        // 写计算sql
        tenv.executeSql(
                "insert into tfc_overview_u_m_doris  " +
                        "with tmp as (                                                                       " +
                        " select                                                                              " +
                        "     gps_province                                                                    " +
                        "     ,gps_city                                                                       " +
                        "     ,gps_region                                                                     " +
                        "     ,release_channel                                                                " +
                        "     ,device_type                                                                    " +
                        "     ,page_type                                                                      " +
                        "     ,page_service                                                                   " +
                        "     ,IF(DATE_FORMAT(register_time,'yyyy-MM-dd') = CURRENT_DATE ,1,0) AS is_newuser  " +
                        " 	,user_id                                                                          " +
                        " 	,session_id                                                                       " +
                        " 	,rt                                                                               " +
                        " from dwd_kafka                                                                      " +
                        " where event_id = 'page_load'                                                        " +
                        " )                                                                                   " +
                        "                                                                                     " +
                        " select                                                                              " +
                        "     to_date(date_format(window_end,'yyyy-MM-dd')) AS dt                                      " +
                        "     ,time_trunc(cast(window_end as string),60)  AS  time_60m                                        " +
                        "     ,time_trunc(cast(window_end as string),30)  AS  time_30m                                        " +
                        "     ,time_trunc(cast(window_end as string),10)  AS  time_10m                                        " +
                        "     ,time_trunc(cast(window_end as string),5)  AS  time_5m                                          " +
                        "     ,gps_province                                                                   " +
                        "     ,gps_city                                                                       " +
                        "     ,gps_region                                                                     " +
                        "     ,release_channel                                                                " +
                        "     ,device_type                                                                    " +
                        "     ,page_type                                                                      " +
                        "     ,page_service                                                                   " +
                        "     ,is_newuser                                                                     " +
                        " 	  ,count(1) as pv_cnt                                                             " +
                        " 	  ,cast(user_id as int) as uv_cnt                                                " +
                        " 	  ,session_id  as session_cnt                                                                 " +
                        " from TABLE(                                                                         " +
                        "   TUMBLE(TABLE tmp, DESCRIPTOR(rt), INTERVAL '1' MINUTE)                            " +
                        " )                                                                                   " +
                        " group by                                                                            " +
                        "     window_start                                                                    " +
                        "     ,window_end	                                                                  " +
                        "     ,gps_province                                                                   " +
                        "     ,gps_city                                                                       " +
                        "     ,gps_region                                                                     " +
                        "     ,release_channel                                                                " +
                        "     ,device_type                                                                    " +
                        "     ,page_type                                                                      " +
                        "     ,page_service                                                                   " +
                        "     ,is_newuser                                                                     " +
                        " 	,user_id                                                                          " +
                        " 	,session_id                                                                       "
        );


    }


}
